package cc.xpcas.raft.network.jvm;

import java.util.*;
import java.util.concurrent.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import lombok.Data;
import lombok.Getter;

/**
 * @author xp
 */
public class JvmHub {

    private static final Logger LOG = LoggerFactory.getLogger(JvmHub.class);

    @Getter
    private static final JvmHub instance = new JvmHub();

    private JvmHub() {
    }

    @Data
    private static class Node {

        private final BlockingQueue<JvmPacket> queue = new LinkedBlockingQueue<>();
    }

    private final Map<JvmAddress, Node> nodeMap = new ConcurrentHashMap<>();

    public void addListener(JvmAddress address) {
        nodeMap.put(address, new Node());
    }

    public void removeListener(JvmAddress address) {
        nodeMap.remove(address);
    }

    public void sendTo(JvmAddress src, JvmAddress dest, byte[] bytes) throws InterruptedException {
        Node node = nodeMap.get(dest);
        if (node == null) {
            if (LOG.isErrorEnabled()) {
                LOG.error(String.format("node not found: %s", System.identityHashCode(dest)));
            }
            return;
        }
        JvmPacket packet = new JvmPacket(src, bytes);
        node.getQueue().put(packet);
    }

    public JvmPacket receiveFrom(JvmAddress address) throws InterruptedException {
        Node node = nodeMap.get(address);
        if (node == null) {
            if (LOG.isErrorEnabled()) {
                LOG.error(String.format("node not found: %s", System.identityHashCode(address)));
            }
            return null;
        }
        return node.getQueue().take();
    }
}
